package com.train.order.mq.consumer;

import cn.hutool.core.bean.BeanUtil;
import com.train.common.constants.KeyConstants;
import com.train.common.constants.mq.MqGroupConstants;
import com.train.common.constants.mq.MqTopicConstants;
import com.train.common.utils.convert.JsonConvertUtils;
import com.train.common.utils.date.LocalDateTimeUtils;
import com.train.common.utils.exception.ExceptionUtils;
import com.train.order.dto.confirm.ConfirmOrderEventDTO;
import com.train.order.enums.impl.ConfirmOrderStatusEnums;
import com.train.order.event.confirmorder.ConfirmOrderEventBody;
import com.train.order.event.confirmorder.ConfirmOrderEventHeader;
import com.train.order.service.ConfirmOrderService;
import com.train.redis.core.support.RedissonSupport;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.jboss.logging.MDC;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@RocketMQMessageListener(topic = MqTopicConstants.ORDER_CONFIRM_TOPIC,
        consumerGroup = MqGroupConstants.ORDER_CONFIRM_CONSUMER_GROUP)
public class ConfirmOrderConsumer implements RocketMQListener<MessageExt> {

    @Autowired
    private ConfirmOrderService confirmOrderService;

    @Autowired
    private RedissonSupport redissonSupport;

    @Override
    public void onMessage(MessageExt messageExt) {
        Map<String, String> headers = messageExt.getProperties();
        ConfirmOrderEventHeader eventHeader = JsonConvertUtils.convert(headers, ConfirmOrderEventHeader.class);
        ConfirmOrderEventBody eventBody = JsonConvertUtils.convert(messageExt.getBody(), ConfirmOrderEventBody.class);
        Long orderId = eventHeader.getBizId();
        RLock lock = redissonSupport.getLock(KeyConstants.MQ_CONFIRM_ORDER_LOCK_KEY, orderId);
        try {
            String messageTime = LocalDateTimeUtils.format(eventHeader.getMessageTime(), LocalDateTimeUtils.DATE_TIME_FORMAT);

            //给该笔订单id设置分布式锁, 防止重复消费
            if (lock.tryLock(0L, TimeUnit.SECONDS)) {
                log.info("消费者开始执行订单确认, 订单号:{}, 时间:{}, 发送者:{}", orderId, messageTime, eventHeader.getMessageSource());

                //设置日志id跟踪
                MDC.put("LOG_ID", eventHeader.getLogId());

                //执行订单确认主流程
                confirmOrderService.confirmProcess(BeanUtil.copyProperties(eventBody, ConfirmOrderEventDTO.class));
            }
        } catch (Exception e) {
            //判断消费的次数是否大于或等于指定重试次数
            if (eventHeader.getRetryCount() != null && messageExt.getReconsumeTimes() >= eventHeader.getRetryCount()) {
                log.error("订单确认主流程处理失败, 原因: {}", ExceptionUtils.getExpMsg(e));
                //更新订单状态
                confirmOrderService.updateConfirmOrderStatus(orderId, ConfirmOrderStatusEnums.FAILURE.getCode());
                return;
            }
            throw new RuntimeException(e);
        } finally {
            redissonSupport.unlock(lock); //释放锁
        }
    }
}
